package com.tyf.common.zk.lock;

import com.tyf.common.exception.KeeperException;
import com.tyf.common.zk.ZKTemplete;
import com.tyf.common.zk.ZKTemplete;
import com.tyf.common.zk.listener.KeeperChildListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.NodeExistsException;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Thread safe
 *
 * @author huangdou
 * @version 0.0.1
 * @at 2016年12月5日下午3:02:52
 * 此类为zookeeper的分布式锁功能
 */
public class KeeperMutexLock implements Lock {
    private String name;

    private String lockPath;


    private ZKTemplete templete;

    private Lock innerLock = new ReentrantLock();
    static final String LOCK_ROOT = "/lock_root";

    private ThreadLocal<String> currentNodePath = new ThreadLocal<String>();
    private ThreadLocal<Integer> lockCount = new ThreadLocal<Integer>();
    private Map<String, Semaphore> waitThreadMap = new ConcurrentHashMap<String, Semaphore>();

    private void innerLocked() {
        innerLock.lock();
    }

    private void innerUnlocked() {
        innerLock.unlock();
    }

    private synchronized void init() {

        if (name == null || "".equals(name.trim()) || name.contains("/")) {
            throw new IllegalArgumentException(String.format("path can not be %s", name));
        }
        lockPath = LOCK_ROOT + "/" + name;
        if (!templete.exist(lockPath)) {
            try {
                templete.createWtihParent(lockPath);
            } catch (KeeperException e) {
                if (!(e.getProto() instanceof NodeExistsException)) {
                    throw e;
                }
            }
        }
        templete.listenChild(lockPath, new MutexLockKeeperChildListener(this));
    }

    public KeeperMutexLock(String name, ZKTemplete templete) {
        super();
        this.name = name;
        this.templete = templete;
        init();
    }

    public static Lock getLock(String name, ZKTemplete templete) {
        return new KeeperMutexLock(name, templete);
    }

    @Override
    public void lock() {
        if (lockCount.get() != null && lockCount.get() > 0) {
            lockCount.set(lockCount.get() + 1);
            return;
        }
        Semaphore semaphoreCurrentThread = new Semaphore(0);
        innerLock.lock();
        try {
            String returnPath = templete.create(lockPath + "/", "".getBytes(), CreateMode.EPHEMERAL_SEQUENTIAL);
            String[] array = returnPath.split("/");
            currentNodePath.set(returnPath);
            waitThreadMap.put(array[array.length - 1], semaphoreCurrentThread);
        } finally {
            innerLock.unlock();
        }
        try {
            semaphoreCurrentThread.acquire();
            lockCount.set(1);
        } catch (InterruptedException e) {
            templete.delete(currentNodePath.get());
            try {
                throw e;
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        this.lock();
    }


    @Override
    public boolean tryLock() {
        try {
            return tryLock(0, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            return false;
        }
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        if (lockCount.get() != null && lockCount.get() > 0) {
            lockCount.set(lockCount.get() + 1);
            return true;
        }
        Semaphore semaphoreCurrentThread = new Semaphore(0);
        innerLock.lock();
        try {
            String returnPath = templete.create(lockPath + "/", "".getBytes(), CreateMode.EPHEMERAL_SEQUENTIAL);
            String[] array = returnPath.split("/");
            currentNodePath.set(returnPath);
            waitThreadMap.put(array[array.length - 1], semaphoreCurrentThread);
        } finally {
            innerLock.unlock();
        }
        boolean isLocked = false;
        if (time > 0) {
            try {
                isLocked = semaphoreCurrentThread.tryAcquire(1, time, unit);
            } catch (InterruptedException e) {
                templete.delete(currentNodePath.get());
                throw e;
            }

        } else {
            isLocked = semaphoreCurrentThread.tryAcquire();
        }
        if (!isLocked) {
            templete.delete(currentNodePath.get());
        } else {
            lockCount.set(1);
        }
        return isLocked;
    }

    @Override
    public void unlock() {
        if (lockCount.get() != null) {
            if (lockCount.get() > 1) {
                lockCount.set(lockCount.get() - 1);
            } else if (lockCount.get() == 1) {
                lockCount.set(null);
                templete.delete(currentNodePath.get());
            }
            return;
        }
        throw new IllegalMonitorStateException();
    }

    @Override
    public Condition newCondition() {
        throw new KeeperException("newCondition is not yet supported!");
    }

    class MutexLockKeeperChildListener implements KeeperChildListener {

        KeeperMutexLock lock;

        public MutexLockKeeperChildListener(KeeperMutexLock lock) {
            super();
            this.lock = lock;
        }

        @Override
        public void onChild(String parent, List<String> subs) {
            innerLocked();
            try {
                for (Entry<String, Semaphore> entry : waitThreadMap.entrySet()) {
                    if (subs.contains(entry.getKey())) {
                        Collections.sort(subs, new Comparator<String>() {
                            @Override
                            public int compare(String o1, String o2) {
                                return Integer.parseInt(o1) - Integer.parseInt(o2);
                            }
                        });
                        if (entry.getKey().equals(subs.get(0))) {
                            entry.getValue().release();
                            waitThreadMap.remove(entry.getKey());
                        }
                    }
                }
            } finally {
                innerUnlocked();
            }
        }

        @Override
        public void onParentDelete(String path) {
        }
    }
}
